package com.markyao.recos.raft.raftRole;

import com.markyao.recos.regist.RecosRegistConfig;
import com.markyao.vo.Result;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j(topic = "m.RaftRoleImpl")
public class RaftRoleImpl extends AbstractRaftRole{


    private RestTemplate restTemplate;


    private AtomicInteger isElection;

    private AtomicInteger sendThreshold;

    private AtomicInteger recvThreshold;

    private final Set<String>adsSet=new HashSet<>();

    private RecosRegistConfig recosRegistConfig;

    public RaftRoleImpl(String host, int port, String username, String password,
                        RestTemplate restTemplate, AtomicInteger isElection,
                        AtomicInteger sendThreshold,AtomicInteger recvThreshold,RecosRegistConfig recosRegistConfig) {
        super(host, port,username,password);
        SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
        factory.setConnectTimeout(60*1000);
        factory.setReadTimeout(60*1000);
        this.restTemplate=new RestTemplate(factory);
        this.isElection=isElection;
        this.sendThreshold=sendThreshold;
        this.recvThreshold=recvThreshold;
        this.recosRegistConfig=recosRegistConfig;
    }

    /**
     * 选举: 广播投票信息
     */
    @Override
    public void election() {
        String me=this.host+":"+this.port;
        for (String ads : this.groupsAds) {
            if (adsSet.contains(ads)){
                continue;
            }
            long startTime = System.currentTimeMillis();
            MultiValueMap<String, String> headrs = getTokenHeadersMap(ads);
            if (headrs==null){
                log.error("与目标recos尚未建立连接! {}",ads);
                continue;
            }
            HttpEntity httpEntity=new HttpEntity(headrs);

            String url=ads + "/recos/raft/vote?username=" + this.username + "&password=" + this.password+"&startTime=";
            ResponseEntity<Result> responseEntity = null;
            try {
                responseEntity = restTemplate.exchange(url+startTime+"&targetAddress="+me,HttpMethod.GET,httpEntity, Result.class);
            } catch (ResourceAccessException e) {
                if (responseEntity==null){
                    //TODO 说明这条连接有问题,可能是目标主机断了，这个时候需要把safeScore往下调整
                    e.printStackTrace();
                    resize(ads);
                    adsSet.add(ads);
                    continue;
                }
            }
            Result result =  responseEntity.getBody();
            if (result!=null && result.getSuccess()) {
                int votes= (int) result.getData();
                int score = this.score.get();
                this.score.compareAndSet(score,score+votes);
            }else {
                resize(ads);
            }
        }
        if (this.score.get()>=this.safeScore){
            //TODO 成功当选,需要关停竞选周期
            log.info("成功当选leader! {}",this.host+":"+this.port);
            this.isElection.set(0);
            shiftLeader();
            startSendHeartBeat();
            recosRegistConfig.setLeader(true);
        }else {
            //TODO 如果当选失败,需要把自己设置成追随者,结束竞选周期
            this.isElection.set(0);
            shiftFollower();
        }
    }

    @Override
    protected void startSendHeartBeat() {
        sendHeartBeat0();
        SendHeartThread sendHeartThread = new SendHeartThread();
        sendHeartThread.start();
    }


    private synchronized void resize(String ads) {
        if (!adsSet.contains(ads)){
            adsSet.add(ads);
            log.error("目标{} recos连接不通",ads);
            this.aliveSize=Math.max(1,aliveSize-1);
            safeScore=(aliveSize/2)+1;
            log.info("重调整safeScore: {}",safeScore);
        }
    }

    @Override
    public void connenct2Recos() {
        String me=this.host+":"+this.port;
        for (String ads : this.groupsAds){
            String url=ads + "/auth0?username=" + this.username + "&password=" + this.password+"&targetAddress="+me;
            ResponseEntity<Result> responseEntity = null;
            int retry=5;
            int waitSec=10;
            while (retry>0 && responseEntity==null){
                try {
                    responseEntity = restTemplate.getForEntity(url, Result.class);
                } catch (ResourceAccessException e) {
                    if (responseEntity==null){
                        log.info("Retrying connect to target recos {}",ads);
                        try {
                            Thread.sleep(waitSec*1000);
                        } catch (InterruptedException ex) {
                            ex.printStackTrace();
                        }
                        retry--;
                        continue;
                    }
                }
                if (responseEntity==null){
                    log.error("目标{} recos连接不通",ads);
                    retry--;
                    continue;
                }else {
                    Result result = responseEntity.getBody();
                    if (result.getSuccess()){
                        String token = result.getData().toString();
                        this.tokenMap.put(ads,token);
                        this.sendThreshold.decrementAndGet();
                        log.info("连接目标recos成功! {}",ads);
                    }else {
                        log.error("目标recos: {} 消息:{}",ads,result.getMsg());
                    }
                }
            }

        }
        //等待接收阈值和发送阈值达到平衡..
        int waitShortSec=10;
        int cnt=30;
        boolean ok=false;
        while (cnt>0){
            try {
                if (recvThreshold.get() == sendThreshold.get()){
                    ok=true;
                    break;
                }
                Thread.sleep(waitShortSec*1000);
                cnt--;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        if (!ok){
            log.error("发送包{} 接收包{} ,两方通信失败,初始化失败.",sendThreshold,recvThreshold);
            return;
        }
        recosRegistConfig.setTokenMap(tokenMap);
        log.info("发送包{} 接收包{}  两方通信初始化成功!",sendThreshold,recvThreshold);
        this.isOnload=false;
    }

    @Override
    public boolean askLeader(String host, int port) {
        String url=host+":"+port+"/recos/raft/askLeader";
        try {
            ResponseEntity<Result> response = restTemplate.exchange(url, HttpMethod.GET, null, Result.class);
            if (response!=null && response.getBody()!=null && response.getBody().getSuccess()){
                return true;
            }
        } catch (Exception e) {
            return false;
        }
        return false;
    }

    @Override
    public synchronized Result vote(Long startTime,String targetAddress) {
        if (isOnload){
            return Result.success(null,0);
        }
        if (this.roleState==2){
            log.error("竞争冲突");
            Integer boom = boomState.getIfPresent("boom");
            try {
                if (boom==null || boom !=1){
                    log.info("放入cache");
                    boomCache();
                }
            } catch (Exception e) {
                boomCache();
            }
            return Result.success(null,0);
        }
        long currentTimeMillis = System.currentTimeMillis();
        if (currentTimeMillis>=startTime+electionTimeWindow){
            log.info("已经过了竞选周期!");
            isElection.set(0);
            return Result.success(null,0);
        }
        if (this.vote){
            log.info("已投过票了!");
            return Result.success(null,0);
        }
        this.vote=true;
        log.info("投{}一票!",targetAddress);
        //开始一个计时器,确保竞选周期能在最坏的情况下结束
        ElectionTimner electionTimner=new ElectionTimner(startTime);
        electionTimner.start();
        return Result.success(null,1);
    }

    private void boomCache() {
        isBoom=true;
        Random rd=new Random();
        int rdn = rd.nextInt(64);
        int expire =rdn * 50;
        log.info("boomCache expire Time: {}",expire);
        boomState.policy().expireVariably().ifPresent(p->p.put("boom",1,expire, TimeUnit.MILLISECONDS));
    }

    /**
     * 第一次发送心跳
     */
    private void sendHeartBeat0() {
        sendHeartBeat(1);
    }

    /**
     * ret为1表示第一次发送心跳，除了基本的心跳激活还需要本机leader的host+port
     * 为0时则是普通的发送心跳
     * @param ret
     */
    @Override
    public void sendHeartBeat(int ret) {
        for (String ads : groupsAds) {
            if (adsSet.contains(ads)){
                continue;
            }
            MultiValueMap<String, String> headrs = getTokenHeadersMap(ads);
            if (ret==1){
                Map<String,Object>body=new HashMap<>();
                body.put("host",this.host);
                body.put("port",this.port);
                HttpEntity httpEntity=new HttpEntity(body,headrs);
                String url=ads + "/recos/raft/recvHeartBeat0";
                recvHeartBeatHTTP(httpEntity, url, ads);
            }else {
                HttpEntity httpEntity=new HttpEntity(null,headrs);
                String url=ads + "/recos/raft/recvHeartBeat";
                recvHeartBeatHTTP(httpEntity,url,ads);
            }
        }
    }
    private ResponseEntity<Result> recvHeartBeatHTTP(HttpEntity entity, String url,String ads){
        ResponseEntity<Result> response= null;
        try {
            response = restTemplate.exchange(url, HttpMethod.POST, entity, Result.class);
        } catch (RestClientException e) {
            if (response==null || (response!=null && response.getBody()==null)){
                adsSet.add(ads);
                log.error("发现有recos节点坏了!! {}",ads);
            }
        }
        return response;
    }
    private MultiValueMap<String, String> getTokenHeadersMap(String ads) {
        if (!tokenMap.containsKey(ads)){
            return null;
        }
        MultiValueMap<String,String>headrs=new LinkedMultiValueMap<>();
        headrs.add("Authorization",tokenMap.get(ads));
        return headrs;
    }

    @Override
    public void recvHeartBeat() {

    }

    /**
     * 竞选周期兜底
     */
    class ElectionTimner extends Thread{
        final long startTime;


        public ElectionTimner(long startTime) {
            this.startTime = startTime;
        }

        @Override
        public void run() {
            long currentTimeMillis = System.currentTimeMillis();
            try {
                Thread.sleep(electionTimeWindow-(currentTimeMillis-startTime));
                isElection.set(0);//停止竞选周期
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 发送心跳线程： 每150ms发送一次心跳
     */
    class SendHeartThread extends Thread{
        private final int T=150;


        @Override
        public void run() {
            log.info("开始向follower广播心跳.");
            while (true){
                sendHeartBeat(0);
                try {
                    Thread.sleep(T);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
